package com.video.chat.websocket;
import com.alibaba.fastjson2.JSON;
import com.video.chat.domain.entity.MessageEntity;
import com.video.chat.domain.entity.SystemMsgEntity;
import com.video.chat.domain.pojo.Message;
import com.video.chat.domain.pojo.ResultMessage;
import com.video.chat.service.MessageService;
import com.video.chat.utils.MessageUtils;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;

import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

/**
 * WebSocket服务
 */
@Component
@ServerEndpoint("/chat/{userId}")
@RequiredArgsConstructor
public class WebSocketServer   {
    private static ApplicationContext applicationContext;
    private static final String ONLINE_USERS_KEY = "online_users";
    public static void setApplicationContext(ApplicationContext context) {
        applicationContext = context;
    }



    // 存储在线的用户，key -> 用户id, value -> session对象
    private static Map<String, Session> onLineUsers = new ConcurrentHashMap<>();
    private static Map<Session, String> onLineUsers00 = new ConcurrentHashMap<>();
    // 缓存未能立即转发的消息
    private Map<String, Queue<Message>> messageQueues = new HashMap<>();
    // 存储待批量存储的消息
    private Map<String, List<Message>> messageBuffer = new ConcurrentHashMap<>();
    // 定期保存消息的线程
    private ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();


    // 初始化时启动定期保存任务
//  s
    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        System.out.println("客户端：" + userId + " 建立连接");
        onLineUsers.put(userId, session);
        onLineUsers00.put(session, userId);
//        UserClient bean = applicationContext.getBean(UserClient.class);
//        AppUser user = bean.info();
//        System.out.println("用户"+user);
        // 将用户 ID 添加到 Redis 的 List 中
        RedisTemplate redisTemplate = (RedisTemplate) applicationContext.getBean("redisTemplate");
        redisTemplate.opsForSet().add(ONLINE_USERS_KEY, userId);
        // 通知所有用户，该用户上线了
        String message = MessageUtils.getMessage(false, null, "有人上线了");
        sendToAllClient(message);
    }

    /**
     * 收到客户端消息后调用的方法
     *
     */
    @OnMessage
    public void onMessage(Session session,String message)  {
        String formName = onLineUsers00.get(session);
        Message msg = JSON.parseObject(message, Message.class);
        String toName = msg.getToName();
        String content = msg.getMessage();
        System.out.println("接受到的消息"+ message);
        //获取当前时间戳
        long currentTimeMillis = System.currentTimeMillis();
        handleMessage(toName,content,formName,currentTimeMillis);
    }

    /**
     * 连接关闭调用的方法
     *
     * @param userId
     */
    @OnClose
    public void onClose(@PathParam("userId") String userId) {
        System.out.println("断开连接"+userId);
        RedisTemplate redisTemplate = (RedisTemplate) applicationContext.getBean("redisTemplate");
        redisTemplate.opsForSet().remove(ONLINE_USERS_KEY, userId);
        // 通知所有用户，该用户下线了
        String message = MessageUtils.getMessage(false, null, "有人下线了");
        sendToAllClient(message);
    }

    /**
     * 群发
     *
     * @param message
     */
    public void sendToAllClient(String message) {
        Collection<Session> sessions = onLineUsers.values();
        System.out.println("当前在线用户：" + sessions);
        for (Session session : sessions) {
            try {
                // 服务器向客户端发送消息
                session.getAsyncRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    /**
     * 系统消息群发
     */
    public void sendSystemMessageToAllClient(SystemMsgEntity message) {
        String systemMsg = MessageUtils.getMessage(true, "系统", message);
        Collection<Session> sessions = onLineUsers.values();
        System.out.println("当前在线用户：" + sessions);
        for (Session session : sessions) {
            try {
                // 服务器向客户端发送消息
                session.getAsyncRemote().sendText(systemMsg);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 处理浏览器发送的消息 -转发
     * @param toName
     * @param content
     */
    private void handleMessage(String toName, String content,String formName,Long createTime) {
        // 创建消息对象
        Message message = new Message(toName,formName,createTime, content);

        saveMessagesToDatabase(toName,message);
        // 尝试直接转发消息
        if (sendMessageDirectly(toName, message)) {
            return;
        }

        // 目标用户不在线,将消息添加到队列
        Queue<Message> queue = messageQueues.computeIfAbsent(toName, k -> new LinkedList<>());
        queue.offer(message);


        // 启动定期任务,检查并重试消息队列
        scheduleRetryTask();
    }


    /**
     * 尝试直接将消息发送给目标用户
     * @param toName 消息接收者
     * @param message 消息对象
     * @return 是否发送成功
     */
    private boolean sendMessageDirectly(String toName, Message message) {
        Session targetSession = onLineUsers.get(toName);
        if (targetSession != null && targetSession.isOpen()) {
            try {
                ResultMessage resultMessage = new ResultMessage("1",false,message.getFromName(),message.getMessage(),toName);
                targetSession.getBasicRemote().sendText(JSON.toJSONString(resultMessage));
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            return true;
        }
        return false;
    }


    /**
     * 启动定期任务,检查并重试消息队列
     */
    private void scheduleRetryTask() {

        // 您可以使用Spring的任务调度功能或其他工具实现这个定期任务
        new Thread(() -> {
            while (!Thread.interrupted()) {
                try {
                    Thread.sleep(5000); // 每 5 秒检查一次消息队列
                    retryMessageQueues();
                } catch (InterruptedException e) {
                    break;
                }
            }
        }).start();
    }


    /**
     * 重试消息队列中的消息
     */
    private void retryMessageQueues() {
        for (Map.Entry<String, Queue<Message>> entry : messageQueues.entrySet()) {
            String toName = entry.getKey();
            Queue<Message> queue = entry.getValue();


            while (!queue.isEmpty()) {
                Message message = queue.poll();

                if (sendMessageDirectly(toName, message)) {
                    System.out.println("Resent message to: " + toName);
                } else {
                    // 如果重试失败,将消息重新添加到队列头部
                    queue.offer(message);
                    break;
                }
            }
        }
    }

    /**
     * 定期将缓存中的消息批量保存到数据库
     */
//    private void flushMessageBuffer() {
//        for (Map.Entry<String, List<Message>> entry : messageBuffer.entrySet()) {
//            String toName = entry.getKey();
//            List<Message> buffer = entry.getValue();
//            if (!buffer.isEmpty()) {
//                saveMessagesToDatabase(toName, buffer);
//                buffer.clear(); // 清空缓存
//            }
//        }
//    }

    /**
     *  将消息列表保存到数据库的方法,您需要根据实际情况实现
     * @param toName
     * @param message
     */
    private void saveMessagesToDatabase(String toName,Message message) {
        MessageService bean = applicationContext.getBean(MessageService.class);
        MessageEntity entity = new MessageEntity();
        entity.setReceiverId(Long.valueOf(toName));
        entity.setIsRead("1");
        entity.setMessageType("text");
        entity.setSenderId(Long.valueOf(message.getFromName()));
        entity.setContent(message.getMessage());
        entity.setCreatedTime(message.getCreateTime());
        bean.saveMessage(entity);
    }
}
